package com.wan.admin;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Preconditions;
import org.apache.flume.Context;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

/**
 * Source工具类
 */
public class RocketMQSourceUtil {

    /**
     * Topic配置项,如：a1.sources.r1.topic=TestTopic
     */
    public static final String TOPIC_CONFIG = "topic";
    public static final String TOPIC_DEFAULT = "FLUME_TOPIC";

    /**
     * Tags配置项,如：a1.sources.r1.tags=Tag1,Tag2
     */
    public static final String TAGS_CONFIG = "tags";
    public static final String TAGS_DEFAULT = "FLUME_TAG";

    /**
     * Topic头部名称配置项,如: a1.sources.r1.topicHeaderName=name
     */
    public static final String TOPIC_HEADER_NAME_CONFIG = "topicHeaderName";
    public static final String TOPIC_HEADER_NAME_DEFAULT = "topic";

    /**
     * Tags头部名称配置项,如: a1.sources.r1.tagsHeaderName=name
     */
    public static final String TAGS_HEADER_NAME_CONFIG = "tagsHeaderName";
    public static final String TAGS_HEADER_NAME_DEFAULT = "tags";

    /**
     * 一次最多拉取条数配置项,如：a1.sources.r1.batchSize=150
     */
    public static final String BATCH_SIZE_CONFIG = "batchSize";
    public static final int BATCH_SIZE_DEFAULT = 32;

    /**
     * Consumer分组配置项,如：a1.sources.r1.consumerGroup=please_rename_unique_group_name
     */
    public static final String CONSUMER_GROUP_CONFIG = "consumerGroup";
    public static final String CONSUMER_GROUP_DEFAULT = "FLUME_CONSUMER_GROUP";

    /**
     * Namesrv地址配置项,如：a1.sinks.s1.nameserver=localhost:9876
     */
    public static final String NAMESERVER_CONFIG = "nameserver";

    /**
     * 订阅方式配置项,如：a1.sources.r1.messageModel=BROADCASTING
     */
    public static final String MESSAGE_MODEL_CONFIG = "messageModel";
    public static final String MESSAGE_MODEL_DEFAULT = "BROADCASTING";

    /**
     * 自定义配置
     */
    public static final String P1_CONFIG = "p1";
    public static final String P1_DEFAULT = "appId";
    public static final String P2_CONFIG = "p2";
    public static final String P2_DEFAULT = "createTime";

    /**
     * 是否输出自定义日志
     */
    public static final String IS_PRINT_LOG = "isPrintLog";


    /**
     * 获取消费者接口对象(拉取消息)
     * @param context 内容
     * @return 消息拉取对象
     */
    public static DefaultMQPullConsumer getConsumer(Context context) {
        //获取消费者组名
        final String consumerGroup = context.getString(CONSUMER_GROUP_CONFIG, CONSUMER_GROUP_DEFAULT);
        //获取消费者组名
        final Boolean isPrintLog = context.getBoolean(IS_PRINT_LOG, true);
        //获取RocketMQ地址
        final String namesrvAddr = Preconditions.checkNotNull(context.getString(NAMESERVER_CONFIG), "必须指定RocketMQ地址, 例如: a1.sources.r1.nameserver=127.0.0.1:9876");
        //使用GroupName来初始化默认拉取消费者
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(consumerGroup);
        if (isPrintLog){
            System.out.println("消费者创建:"+ JSON.toJSONString(consumer));
        }
        //设置地址,消息订阅方式
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setMessageModel( MessageModel.valueOf(context.getString(MESSAGE_MODEL_CONFIG, MESSAGE_MODEL_DEFAULT)));
        if (isPrintLog){
            System.out.println("消费者设置地址与订阅:"+ JSON.toJSONString(consumer));
        }
        //返回消费者
        return consumer;
    }

}
